package com.glodon.paas.framework.registry.client.web;

import com.glodon.paas.framework.registry.client.util.EnvironmentUtils;
import com.glodon.paas.framework.registry.core.status.ClientStatus;
import com.glodon.paas.framework.registry.core.type.ClientBasic;
import com.glodon.paas.framework.registry.core.type.ServiceBasic;
import com.glodon.paas.framework.registry.core.zk.SessionStateProcessor;
import com.glodon.paas.framework.registry.core.zk.SessionStateWatcher;
import com.glodon.paas.framework.registry.core.zk.ZooKeeperUtils;
import org.apache.commons.lang.SerializationUtils;
import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.servlet.ServletContext;
import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import static com.google.common.base.Strings.isNullOrEmpty;

/**
 * SRDS client bootstrap listener, which bootstrap the SRDS client.
 *
 * @author Don Li
 */
public class ClientBootstrapListener implements ServletContextListener {

    public static final String REGISTRY_SERVICES_PREFIX = "/registry/services/";

    public static final String CLIENT_ZOOKEEPER = "srds_client_zookeeper";
    public static final String CLIENT_PATH = "srds_client_path";
    public static final String CLIENT_STATUS = "srds_client_status";

    public static final int DEFAULT_TIMEOUT = 30;
    public static final int DEFAULT_RETRY = 3;

    private static final Logger LOGGER = LoggerFactory.getLogger(ClientBootstrapListener.class);

    private ServletContext context;

    private boolean srdsEnabled;
    private String connectString;
    private int sessionTimeout;
    private String serviceId;
    private String serviceName;
    private String healthCheck;
    private int retries;

    private String ip;
    private int port;

    private ZooKeeper zk;

    @Override
    public void contextInitialized(ServletContextEvent sce) {
        context = sce.getServletContext();
        loadProperties();

        if (srdsEnabled) {
            validateProperties();
            zk = ZooKeeperUtils.createZooKeeper(connectString, sessionTimeout, createSessionStateWatcher(), retries);
            context.setAttribute(CLIENT_ZOOKEEPER, zk);
        }
    }

    @Override
    public void contextDestroyed(ServletContextEvent sce) {
        if (srdsEnabled) {
            context.removeAttribute(CLIENT_ZOOKEEPER);
            try {
                zk.close();
            } catch (InterruptedException e) {
                LOGGER.error("Error disconnecting from zk cluster", e);
            }
        }
    }

    private SessionStateWatcher createSessionStateWatcher() {
        List<Watcher.Event.KeeperState> states = new ArrayList<Watcher.Event.KeeperState>();
        states.add(Watcher.Event.KeeperState.SyncConnected);
        final SessionStateWatcher watcher = new SessionStateWatcher();
        watcher.setRegisteredState(states);
        watcher.setProcessor(new SessionStateProcessor() {
            @Override
            public void process(WatchedEvent event) {
                // TODO if zk client not able to connected to the local zk cluster, then try remote ones
                if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                    int counter = DEFAULT_RETRY;
                    while (counter-- > 0) {
                        try {
                            // if the 1st instance of that service, then create the service node first
                            final String servicePath = REGISTRY_SERVICES_PREFIX + serviceId;
                            if (zk.exists(servicePath, false) == null) {
                                createServiceNode(servicePath);
                            }
                            // create the client node
                            String clientPath = createClientNode(servicePath);
                            context.setAttribute(CLIENT_PATH, clientPath);
                        } catch (Exception e) {
                            if (counter > 0) {
                                continue;
                            }
                            LOGGER.error("Error registering with zk cluster", e);
                            break;
                        }
                        LOGGER.info("host [{}] has been successfully registered as a client of service [{}]", ip, serviceId);
                        break;
                    }
                } else if (event.getState() == Watcher.Event.KeeperState.Expired) {
                    LOGGER.info("ZK session expired, now create a new zk client");
                    zk = ZooKeeperUtils.createZooKeeper(connectString, sessionTimeout, watcher, retries);
                }
            }
        });
        return watcher;
    }

    private void createServiceNode(String servicePath) throws KeeperException, InterruptedException {
        // build the service detail
        ServiceBasic serviceBasic = new ServiceBasic(serviceId, serviceName, healthCheck);
        /*// build the acl for the service node
        List<ACL> serviceAcls = new ArrayList<ACL>();
        serviceAcls.add(new ACL(ZooDefs.Perms.READ, ZooDefs.Ids.ANYONE_ID_UNSAFE)); // allow the world to read
        serviceAcls.add(new ACL(ZooDefs.Perms.ALL, new Id("digest", "298ca04dc839421432477d3eacc53a14")));
        serviceAcls.add(new ACL(ZooDefs.Perms.ALL, new Id("ip", "10.0.0.1")));*/
        // register the service
        zk.create(servicePath, SerializationUtils.serialize(serviceBasic), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    private String createClientNode(String servicePath) throws KeeperException, InterruptedException {
        // create the client node
        final String clientPath = servicePath + "/" + EnvironmentUtils.getInstanceName();
        String realPath = zk.create(clientPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // set the client detail
        String clientId = realPath.substring(realPath.lastIndexOf("/"));
        ClientBasic clientBasic = new ClientBasic(clientId, ClientStatus.STARTING, ip, port);
        zk.setData(realPath, SerializationUtils.serialize(clientBasic), 0);
        return realPath;
    }

    private void loadProperties() {
        String configFile = context.getInitParameter("srdsConfigFile");
        if (isNullOrEmpty(configFile) || !configFile.startsWith("classpath:")) {
            throw new RuntimeException("config file not invalid");
        }
        String path = configFile.substring(10);
        if (isNullOrEmpty(path)) {
            throw new RuntimeException("config file not specified");
        }
        Properties props = new Properties();
        InputStream inputStream = Thread.currentThread().getContextClassLoader().getResourceAsStream(path);
        if (inputStream == null) {
            LOGGER.error("Configuration file {} not found", configFile);
            return;
        }
        try {
            props.load(inputStream);
            srdsEnabled = Boolean.parseBoolean(props.getProperty("feature_flag.srds.enabled", "false"));
            connectString = props.getProperty("srds.zk.connect_string");
            sessionTimeout = Integer.parseInt(props.getProperty("srds.zk.session_timeout"));
            retries = Integer.parseInt(props.getProperty("srds.zk.retries", "" + DEFAULT_RETRY));
            serviceId = props.getProperty("srds.service.id");
            serviceName = props.getProperty("srds.service.name");
            healthCheck = props.getProperty("srds.service.health_check");
            ip = props.getProperty("srds.instance.ip", EnvironmentUtils.getIp("eth0"));
            port = Integer.parseInt(props.getProperty("srds.instance.port"));
        } catch (IOException e) {
            LOGGER.error("Error loading configuration file", e);
        } finally {
            try {
                inputStream.close();
            } catch (IOException e) {
                LOGGER.warn("Unable to close inputstream", e);
            }
        }
    }

    private void validateProperties() {
        if (isNullOrEmpty(connectString)) {
            throw new RuntimeException("Property [srds.zk.connect_string] cannot be empty");
        }
        if (isNullOrEmpty(serviceId)) {
            throw new RuntimeException("Property [srds.service.id] cannot be empty");
        }
        if (isNullOrEmpty(ip)) {
            throw new RuntimeException("Property [srds.instance.ip] cannot be empty");
        }
    }
}
